package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.ExamPaperCourseTestBean;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.util.ClickHouseUtil;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import com.atguigu.edu.realtime.util.TimestampLtz3CompareUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;


public class DwsExamPaperCourseTestWindow {



    public static void main(String[] args) throws Exception {

        // TODO 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment
                .getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2. 检查点相关设置(略)
        // TODO 3. 从 Kafka dwd_trade_order_detail 主题读取订单明细数据
        String topic = "dwd_interaction_exam";
        String groupId = "dws_exam_paper_course_test_window";

        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> source = env.addSource(kafkaConsumer);


        // TODO 4.转换数据结构
        SingleOutputStreamOperator<JSONObject> mappedStream = source.map(JSON::parseObject);

        // TODO 5. 按照 order_detail_id 分组
        KeyedStream<JSONObject, String> keyedStream = mappedStream.keyBy(r -> r.getString("id"));

        // TODO 6. 去重
        SingleOutputStreamOperator<JSONObject> processedStream = keyedStream.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    private ValueState<JSONObject> lastValueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        lastValueState = getRuntimeContext().getState(
                                new ValueStateDescriptor<JSONObject>("last_value_state", JSONObject.class)
                        );
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject lastValue = lastValueState.value();
                        if (lastValue == null) {
                            long currentProcessingTime = ctx.timerService().currentProcessingTime();
                            ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                            lastValueState.update(jsonObj);
                        } else {
                            String lastRowOpTs = lastValue.getString("row_op_ts");
                            String rowOpTs = jsonObj.getString("row_op_ts");
                            if (TimestampLtz3CompareUtil.compare(lastRowOpTs, rowOpTs) <= 0) {
                                lastValueState.update(jsonObj);
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws IOException {
                        JSONObject lastValue = this.lastValueState.value();
                        if (lastValue != null) {
                            out.collect(lastValue);
                        }
                        lastValueState.clear();
                    }
                }
        );
        // TODO 7. 转换数据结构
        SingleOutputStreamOperator<ExamPaperCourseTestBean> javaBeanStream = processedStream.map(
                jsonObj -> {
                    String paperId = jsonObj.getString("paper_id");
                    String examId = jsonObj.getString("id");
                    String userId = jsonObj.getString("user_id");
                    Double score = jsonObj.getDouble("score");
                    Long durationSec = jsonObj.getLong("duration_sec");
                    Long ts = jsonObj.getLong("ts") * 1000L;

                    ExamPaperCourseTestBean examPaperCourseTestBean = ExamPaperCourseTestBean.builder()
                            .paperId(paperId)
                            .userId(userId)
                            .durationSum(durationSec)
                            .examIdSet(new HashSet<String>(
                                    Collections.singleton(examId)
                            ))
                            .scoreSum(score)
                            .ts(ts)
                            .build();
                    return examPaperCourseTestBean;
                }
        );



        // TODO 9. 按照省份 ID 分组
        KeyedStream<ExamPaperCourseTestBean, String> keyedByProIdStream =
                javaBeanStream.keyBy(ExamPaperCourseTestBean::getUserId);

        SingleOutputStreamOperator<ExamPaperCourseTestBean> uuCountDS = keyedByProIdStream.process(
                new KeyedProcessFunction<String, ExamPaperCourseTestBean, ExamPaperCourseTestBean>() {

                    private ValueState<String> lastOrderDt;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        ValueStateDescriptor<String> valueStateDescriptor =
                                new ValueStateDescriptor<>("course_window_last_order_dt", String.class);
                        // 设置状态 ttl
                        valueStateDescriptor.enableTimeToLive(
                                StateTtlConfig
                                        .newBuilder(Time.days(1L))
                                        .updateTtlOnCreateAndWrite()
                                        .build()
                        );
                        lastOrderDt = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(ExamPaperCourseTestBean value, KeyedProcessFunction<String, ExamPaperCourseTestBean, ExamPaperCourseTestBean>.Context ctx, Collector<ExamPaperCourseTestBean> out) throws Exception {
                        Long ts = value.getTs();
                        String curDate = DateFormatUtil.toDate(ts);
                        String lastDt = lastOrderDt.value();
                        if (lastDt == null || lastDt.equals(curDate)) {
                            value.setUserCount(1L);
                            lastOrderDt.update(curDate);
                        }
                        out.collect(value);

                    }


                }
        );

        // TODO 10. 设置水位线
        SingleOutputStreamOperator<ExamPaperCourseTestBean> watermarkDS = uuCountDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<ExamPaperCourseTestBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<ExamPaperCourseTestBean>() {
                                    @Override
                                    public long extractTimestamp(ExamPaperCourseTestBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )
        );

        KeyedStream<ExamPaperCourseTestBean, String> keyedForAggregateStream = watermarkDS.keyBy(
                new KeySelector<ExamPaperCourseTestBean, String>() {

                    @Override
                    public String getKey(ExamPaperCourseTestBean javaBean) throws Exception {
                        return javaBean.getPaperId();
                    }
                }
        );

        // TODO 10. 开窗
        WindowedStream<ExamPaperCourseTestBean, String, TimeWindow> windowDS = keyedForAggregateStream.window(TumblingEventTimeWindows.of(
                org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)
        ));

        // TODO 11. 聚合计算
        SingleOutputStreamOperator<ExamPaperCourseTestBean> reducedStream = windowDS.reduce(
                new ReduceFunction<ExamPaperCourseTestBean>() {
                    @Override
                    public ExamPaperCourseTestBean reduce(ExamPaperCourseTestBean value1, ExamPaperCourseTestBean value2) throws Exception {
                        value1.getExamIdSet().addAll(
                                value2.getExamIdSet()
                        );
                        value1.setScoreSum(
                                value1.getScoreSum() + value2.getScoreSum()
                        );
                        value1.setDurationSum(
                                value1.getDurationSum()+value2.getDurationSum()
                        );
                        value1.setUserCount(
                                value1.getUserCount()+value2.getUserCount()
                        );
                        return value1;
                    }
                },
                new ProcessWindowFunction<ExamPaperCourseTestBean, ExamPaperCourseTestBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<ExamPaperCourseTestBean> elements, Collector<ExamPaperCourseTestBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        for (ExamPaperCourseTestBean element : elements) {
                            element.setStt(stt);
                            element.setEdt(edt);
                            element.setExamCount((long) element.getExamIdSet().size());
                            element.setTs(System.currentTimeMillis());
                            out.collect(element);
                        }
                    }
                }
        );

        SingleOutputStreamOperator<ExamPaperCourseTestBean> withPaperStream = AsyncDataStream.unorderedWait(
                reducedStream,
                new DimAsyncFunction<ExamPaperCourseTestBean>("dim_test_paper".toUpperCase()) {

                    @Override
                    public void join(ExamPaperCourseTestBean javaBean, JSONObject jsonObj) throws Exception {
                        String paperName = jsonObj.getString("paper_title".toUpperCase());
                        String courseId = jsonObj.getString("course_id".toUpperCase());
                        javaBean.setPaperName(paperName);
                        javaBean.setCourseId(courseId);
                    }

                    @Override
                    public String getKey(ExamPaperCourseTestBean javaBean) {
                        return javaBean.getPaperId();
                    }
                },
                60 * 50, TimeUnit.SECONDS
        );

        SingleOutputStreamOperator<ExamPaperCourseTestBean>  fullInfoStream = AsyncDataStream.unorderedWait(
                withPaperStream,
                new DimAsyncFunction<ExamPaperCourseTestBean>("dim_course_info".toUpperCase()) {

                    @Override
                    public void join(ExamPaperCourseTestBean javaBean, JSONObject jsonObj) throws Exception {

                        String courseName = jsonObj.getString("course_name".toUpperCase());

                        javaBean.setCourseName(courseName);
                    }

                    @Override
                    public String getKey(ExamPaperCourseTestBean javaBean) {
                        return javaBean.getCourseId();
                    }
                },
                60 * 50, TimeUnit.SECONDS
        );


        SinkFunction<ExamPaperCourseTestBean> jdbcSink = ClickHouseUtil.<ExamPaperCourseTestBean>getJdbcSink(
                "insert into dws_exam_paper_course_test_window values(?,?,?,?,?,?,?,?,?,?,?,?)"
        );
        fullInfoStream.<ExamPaperCourseTestBean>addSink(jdbcSink);
        fullInfoStream.print(">>>>>>");



        env.execute();
    }

}
